/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.server;

import cn.ac.iie.di.handler.WriteHandler;
import cn.ac.iie.di.util.Monitor;
import cn.ac.iie.di.util.GlobalParas;
import cn.ac.iie.di.kill.KillHandler;
import static cn.ac.iie.di.kill.KillHandler.log;
import cn.ac.iie.di.rocketmq.RocketMqClient;
import org.apache.log4j.Logger;

/**
 *
 * @author YG
 */
public class DataDumpMain {

    public static Logger log = Logger.getLogger(DataDumpMain.class.getName());

    public static void main(String[] args) {
        GlobalParas.init();

//        KillHandler killhandle = new KillHandler();
//        killhandle.registerSignal("TERM");
//        log.info("init the kill handler done ");

        Monitor moni = new Monitor();
        Thread tMoni = new Thread(moni);
        tMoni.setName("Monitor");
        tMoni.start();

        for (int i = 0; i < GlobalParas.writeThreadSize; ++i) {
            Thread twa = new Thread(new WriteHandler());
            twa.setName("WriteHandler-" + i);
            twa.start();
        }

        for (String oriMsgMq : GlobalParas.oriMsgMqs) {
            RocketMqClient rmc = new RocketMqClient(oriMsgMq);
            Thread rmct = new Thread(rmc);
            rmct.setName("RcoketMqClient---" + oriMsgMq);
            GlobalParas.rocketmqclientQueue.offer(rmc);
            rmct.start();
        }
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                log.info("shutting the rockert consumer ... ");
                log.info("GlobalParas.RocketmqclientQueue size is : " + GlobalParas.rocketmqclientQueue.size());
                try {
                    while (!GlobalParas.rocketmqclientQueue.isEmpty()) {
                        RocketMqClient rmc = GlobalParas.rocketmqclientQueue.poll();
                        rmc.shutDownConsumer();
                        log.info("kill a pull thread ");
                        GlobalParas.isShouldExit = true;
                    }
                    log.info("GlobalParas.RocketmqclientQueue size is : ********" + GlobalParas.rocketmqclientQueue.size());

                    while (true) {
                        log.info("GlobalParas.oriMsgPool available size is : " + GlobalParas.oriMsgPool.size());
                        Thread.sleep(1000);
                        if (GlobalParas.oriMsgPool.isEmpty() && GlobalParas.isShouldExit) {
                            GlobalParas.spm.stop();
                            log.info("GlobalParas.spm.stop OK");
                            break;
                        }
                    }
                } catch (InterruptedException ex) {
                    log.error(ex.getMessage(), ex);
                } catch (Exception ex) {
                    log.error(ex.getMessage(), ex);
                }
            }
        });
    }
}
